#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2024/10/22 10:26
# @Author  : 王凯
# @File    : clean_to_task.py
# @Project : scrapy_spider
import concurrent.futures
import datetime
import json

import pandas as pd
import parsel
from loguru import logger

from apps.pdd.pdd.clean import TABLE_NET_PDD_SHOP_INFO
from components.config import WFQ_SOURCE_MYSQL_CONFIG
from utils.db.mysqldb import MysqlDB


class CleanToTask:

    def __init__(self):
        self._to_db = None

    @property
    def to_db(self):
        if self._to_db is None:
            self._to_db = MysqlDB(
                ip=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_IP"],
                port=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_PORT"],
                db=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_DB"],
                user_name=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_NAME"],
                user_pass=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_PASS"],
            )
        return self._to_db

    def is_update_out_state(self, mall_id):
        sql = f"""select mall_id from {TABLE_NET_PDD_SHOP_INFO} where mall_id = '{mall_id}' and owner is not null"""
        res = self.to_db.find(sql, to_json=True)
        if res:
            return True
        else:
            return False

    def update_time_by_mall_id(self, mall_id, out_create_time=None):
        if out_create_time is None:
            return
        sql = f"""select mall_id, out_state, out_create_time from {TABLE_NET_PDD_SHOP_INFO} where mall_id = '{mall_id}' and owner is not null"""
        res = self.to_db.find(sql, to_json=True)
        if res:
            item = res[0]
            if item.get("out_state") == 2:
                if item.get("out_create_time"):
                    # 任务完成不更新
                    return
            item["out_create_time"] = out_create_time
            self.to_db.update_smart(TABLE_NET_PDD_SHOP_INFO, item, f"mall_id = '{mall_id}'")

    @staticmethod
    def get_raw_data(html):
        response = parsel.Selector(html)
        raw_data = response.re_first(r"window\.rawData\s*=\s*(\{.*\});")
        if not raw_data:
            logger.error("未获取到数据")
            return False
        raw_data = json.loads(raw_data)
        return raw_data

    def _parse_goods_from_json_and_save_to_db(self, info_dict, mall_id, update_date):
        res = []
        goods_list = info_dict.get("goods_list") or []
        for goods in goods_list:
            goods_id = goods.get("goods_id")
            goods_name = goods.get("goods_name")
            price = goods.get("price")
            cnt = goods.get("cnt")
            sales_tip = goods.get("sales_tip")
            item = {
                "mall_id": mall_id,
                "goods_id": goods_id,
                "goods_name": goods_name,
                "price": price,
                "sale_num": cnt,
                "sales_tip": sales_tip,
                "update_date": update_date,
                "hd_url": goods.get("hd_url"),
                "market_price": goods.get("market_price"),
                "tag_list": goods.get("tag_list"),
                "normal_price": goods.get("normal_price"),
            }
            res.append(item)
        return res

    def _parse_goods_one_mall(self, df_one_mall_goods):
        if df_one_mall_goods.empty:
            return
        data_update_time = df_one_mall_goods.create_time.max()
        mall_id = df_one_mall_goods["mall_id"].iloc[0]
        df_one_mall_goods_new = df_one_mall_goods[
            ~(  # 异常的数据
                (df_one_mall_goods["response"] == {})
                | (
                    df_one_mall_goods["response"].map(
                        lambda x: "verify_auth_token" in x
                    )
                )
                | (df_one_mall_goods["response"] == {"error_code": 40001})
            )
        ]
        df_one_mall_goods_new.sort_values(by=["id"], ascending=True)
        df_one_mall_goods_new = df_one_mall_goods_new.drop_duplicates(
            subset=["page_no"], keep="last"
        )
        is_all = any(
            [
                True
                for i in df_one_mall_goods_new["response"].apply(
                    lambda x: x.get("has_more", True)
                )
                if i is False
            ]
        )
        is_all_flag = 1 if is_all is True else 0
        if is_all_flag == 1:
            self.update_time_by_mall_id(mall_id, data_update_time)
            self.to_db.update_smart(
                TABLE_NET_PDD_SHOP_INFO,
                {
                    "is_all": is_all_flag,
                    "shop_status": "开店",
                    "now_crawler_page_num": df_one_mall_goods_new.shape[0],
                    "flag": 1,
                    **({"out_state": 2} if self.is_update_out_state(mall_id) else {}),
                },
                f"mall_id = '{mall_id}' and (flag != -1 or flag is null) ",
            )
        else:
            self.update_time_by_mall_id(mall_id, data_update_time)
            if df_one_mall_goods_new.shape[0] >= 10:
                is_all = 1
                self.to_db.update_smart(
                    TABLE_NET_PDD_SHOP_INFO,
                    {
                        "is_all": 1,
                        "shop_status": "开店",
                        "now_crawler_page_num": df_one_mall_goods_new.shape[0],
                        "flag": 1,
                        **(
                            {"out_state": 2}
                            if self.is_update_out_state(mall_id)
                            else {}
                        ),
                    },
                    f"mall_id = '{mall_id}' and (flag != -1 or flag is null) ",
                )
            else:
                self.update_time_by_mall_id(mall_id, data_update_time)
                self.to_db.update_smart(
                    TABLE_NET_PDD_SHOP_INFO,
                    {
                        "is_all": is_all_flag,
                        "now_crawler_page_num": df_one_mall_goods_new.shape[0],
                        "out_state": 3,
                    },
                    f"mall_id = '{mall_id}' and (flag != -1 or flag is null) ",
                )
        logger.info(
            f"店铺 {mall_id} 商品页数 {df_one_mall_goods_new.shape[0]} 店铺是否采集完整 {is_all} "
        )
        results = []
        for _, row in df_one_mall_goods_new.iterrows():
            update_date = str(row["create_time"])[:10]
            # 保存商品详情
            res = self._parse_goods_from_json_and_save_to_db(
                row["response"], mall_id, update_date
            )
            if res:
                results.extend(res)

        if results:
            self.to_db.add_batch_smart(
                "net_pdd_goods_detail",
                results,
                update_columns=[
                    "mall_id",
                    "goods_id",
                    "goods_name",
                    "price",
                    "sale_num",
                    "sales_tip",
                    "hd_url",
                    "market_price",
                    "tag_list",
                    "normal_price",
                ],
            )

    def get_datas(self, mall_ids):
        time_str = datetime.datetime.now().replace(day=1).strftime("%Y-%m-%d")
        mall_ids_str = ",".join([f"'{i}'" for i in mall_ids])
        datas = self.to_db.find(
            f"select * from net_pdd_proxy_log where mall_id in ({mall_ids_str}) and create_time > '{time_str}' and (source = 'app' or source = 'web' or source = 'remote')",
            to_json=True,
        )
        return pd.DataFrame(datas)

    def parse_query_cat_goods(self, df_pro):
        if df_pro.empty:
            return df_pro
        goods_df = df_pro[
            df_pro["base_url"].str.contains("api/turing/mall/query_cat_goods")
        ].copy()
        goods_df.groupby("mall_id", as_index=False).apply(
            lambda df_one: self._parse_goods_one_mall(df_one)
        )
        return df_pro

    def _parse_one_mall(self, df_one_mall):
        mall_id = df_one_mall["mall_id"].iloc[0]
        data_update_time = df_one_mall["create_time"].max()
        df_one_mall = df_one_mall.sort_values(by=["id"], ascending=False)
        shop_name = None
        mall_star = None
        mall_details_label_list = None
        open_time_label = None
        mall_desc = None
        mall_logo = None
        mall_label_volist = None
        mall_rating_text_list = None
        service_score = None
        shop_status = None
        flag = None
        out_state = None
        shop_url = f"https://mobile.yangkeduo.com/mall_page.html?mall_id={mall_id}"
        for _, row in df_one_mall.iterrows():
            data = row["response"]
            if "data" in data:
                if data["data"]:
                    mall_detail_info = data["data"]["store"].get("mallDetailInfo")
                    if mall_detail_info:
                        mall_basic_info = mall_detail_info.get("mallBasicInfo")
                        if mall_basic_info:
                            shop_name = shop_name or mall_basic_info["mallName"]
                            mall_desc = mall_desc or mall_basic_info["mallDesc"]
                            mall_logo = mall_logo or mall_basic_info["mallLogo"]
                        mall_star = mall_star or (
                            mall_detail_info.get("dsr") or {}
                        ).get("mallStar")
                        mall_details_label_list = (
                            mall_details_label_list
                            or mall_detail_info.get("mallDetailsLabelList")
                        )
                        mall_label_volist = mall_label_volist or mall_detail_info.get(
                            "mallLabelVolist"
                        )
                        for k in mall_label_volist or []:
                            if "年老店" in k["desc"]:
                                open_time_label = open_time_label or k["desc"]
                        mall_rating_text_list = mall_rating_text_list or (
                            mall_detail_info.get("dsr") or {}
                        ).get("mallRatingTextList")
                        service_score = service_score or (
                            mall_detail_info.get("dsr") or {}
                        ).get("serviceScore")
                        open_time_label = (
                            open_time_label
                            or (
                                mall_detail_info.get("openTimeLabel") or {"desc": None}
                            )["desc"]
                        )
                    else:
                        mall_basic_info = data["data"]["store"].get("mallBasicInfo")
                        if mall_basic_info:
                            shop_name = shop_name or mall_basic_info.get("mallName")
                            mall_desc = mall_desc or mall_basic_info.get("mallDesc")
                            mall_logo = mall_logo or mall_basic_info.get("mallLogo")
                            mall_label_volist = (
                                mall_label_volist
                                or mall_basic_info.get("mallLabelList")
                            )
                            for k in mall_label_volist or []:
                                if "年老店" in k["desc"]:
                                    open_time_label = open_time_label or k["desc"]
                            if mall_basic_info.get("isOpen") == 0:
                                shop_status = "该店铺已失效"
                                flag = -1
                                out_state = 2
                            else:
                                if mall_basic_info.get("hasGoods") is False:
                                    no_goods_list_tips = (
                                        parsel.Selector(text=data["html"])
                                        .xpath(
                                            '//div[contains(@class, "no-goods-list-tips")]/text()'
                                        )
                                        .get()
                                    )
                                    if no_goods_list_tips:
                                        if no_goods_list_tips == "该店铺正在上传商品":
                                            shop_status = "该店铺正在上传商品"
                                            flag = -1
                                            out_state = 2
                                        if no_goods_list_tips == "该店铺已失效":
                                            shop_status = "该店铺已失效"
                                            flag = -1
                                            out_state = 2
        if shop_status is None:
            shop_status = "开店"
            flag = 1
        if shop_name and mall_id:
            item = {
                "shop_name": shop_name,
                "mall_star": mall_star,
                "mall_details_label_list": mall_details_label_list,
                "shop_url": shop_url,
                "open_time_label": open_time_label,
                "mall_desc": mall_desc,
                "mall_logo": mall_logo,
                "mall_label_volist": mall_label_volist,
                "mall_rating_text_list": mall_rating_text_list,
                "service_score": service_score,
            }
            if shop_status:
                item["shop_status"] = shop_status
            if flag is not None:
                item["flag"] = flag
            if out_state is not None:
                if self.is_update_out_state(mall_id):
                    item["out_state"] = out_state
            logger.info(f"店铺 {mall_id} 店铺信息 {item}")
            self.update_time_by_mall_id(mall_id, data_update_time)
            self.to_db.update_smart(TABLE_NET_PDD_SHOP_INFO, item, f"mall_id = '{mall_id}'")

    def parse_mall_page(self, df_pro):
        if df_pro.empty:
            return df_pro
        mall_df = df_pro[
            df_pro["base_url"].str.contains("/mall_subject.html")
            | df_pro["base_url"].str.contains("/mall_page.html")
        ].copy()
        mall_df_new = mall_df[~mall_df["response"].map(lambda x: x["data"] is None)]
        if not mall_df_new.empty:
            mall_df_new.groupby("mall_id", as_index=False).apply(
                lambda df_one: self._parse_one_mall(df_one)
            )
        return df_pro

    def _parse_one_mall_combination(self, df_one_mall):
        mall_id = df_one_mall["mall_id"].iloc[0]
        df_one_mall.sort_values(by=["id"], ascending=False)
        shop_name = None
        mall_desc = None
        mall_logo = None
        shop_status = None
        flag = None
        shop_url = f"https://mobile.yangkeduo.com/mall_page.html?mall_id={mall_id}"
        for _, row in df_one_mall.iterrows():
            data = row["response"]
            if "result" in data:
                mall_basic_info = data["result"].get("mall_basic_info")
                if mall_basic_info:
                    shop_name = shop_name or mall_basic_info.get("mall_name")
                    mall_desc = mall_desc or mall_basic_info.get("mall_desc")
                    mall_logo = mall_logo or mall_basic_info.get("mall_logo")
                    if mall_basic_info.get("is_open") == 0:
                        shop_status = "该店铺已失效"
                        flag = -1

        if shop_name and mall_id:
            item = {
                "shop_name": shop_name,
                "shop_url": shop_url,
                "mall_desc": mall_desc,
                "mall_logo": mall_logo,
            }
            if shop_status:
                item["shop_status"] = shop_status
            if flag is not None:
                item["flag"] = flag
            logger.info(f"店铺 {mall_id} 店铺信息 {item}")
            self.to_db.update_smart(TABLE_NET_PDD_SHOP_INFO, item, f"mall_id = '{mall_id}'")

    def _parse_one_mall_info(self, df_one_mall):
        mall_id = df_one_mall["mall_id"].iloc[0]
        df_one_mall.sort_values(by=["id"], ascending=False)
        shop_name = None
        mall_desc = None
        shop_status = None
        flag = None
        goods_num = None
        mall_sales = None
        logo = None
        is_open = None
        for _, row in df_one_mall.iterrows():
            mall_info = row["response"]
            shop_name = mall_info.get("mall_name")
            mall_desc = mall_info.get("mall_desc")
            goods_num = mall_info.get("goods_num")
            mall_sales = mall_info.get("mall_sales")
            logo = mall_info.get("logo")
            is_open = mall_info.get("is_open")

        if shop_name and mall_id:
            item = {
                "shop_name": shop_name,
                "mall_goods_num": goods_num,
                "mall_sales": mall_sales,
                "mall_desc": mall_desc,
                "mall_logo": logo,
            }
            if is_open == 0:
                shop_status = "该店铺已失效"
                flag = -1

            if goods_num == 0:
                item['is_all'] = 1
            if shop_status:
                item["shop_status"] = shop_status
            if flag is not None:
                item["flag"] = flag
            logger.info(f"店铺 {mall_id} 店铺信息 {item}")
            self.to_db.update_smart(TABLE_NET_PDD_SHOP_INFO, item, f"mall_id = '{mall_id}'")

    def _parse_one_mall_detail(self, df_one_mall):
        mall_id = df_one_mall["mall_id"].iloc[0]
        df_one_mall.sort_values(by=["id"], ascending=False)
        shop_name = None
        mall_star = None
        mall_details_label_list = None
        open_time_label = None
        mall_desc = None
        mall_logo = None
        mall_label_volist = None
        mall_rating_text_list = None
        service_score = None
        shop_status = None
        flag = None
        shop_url = f"https://mobile.yangkeduo.com/mall_page.html?mall_id={mall_id}"
        for _, row in df_one_mall.iterrows():
            mall_detail_info = row["response"]
            if mall_detail_info:
                mall_basic_info = mall_detail_info.get("mall_basic_info")
                if mall_basic_info:
                    shop_name = shop_name or mall_basic_info.get("mall_name")
                    mall_desc = mall_desc or mall_basic_info.get("mall_desc")
                    mall_logo = mall_logo or mall_basic_info.get("mall_logo")
                mall_star = mall_star or (mall_detail_info.get("dsr") or {}).get(
                    "mall_star"
                )
                mall_details_label_list = (
                    mall_details_label_list
                    or mall_detail_info.get("mall_details_label_list")
                )
                mall_label_volist = mall_label_volist or mall_detail_info.get(
                    "mall_label_volist"
                )
                for k in mall_label_volist or []:
                    if "年老店" in k["desc"]:
                        open_time_label = open_time_label or k["desc"]
                mall_rating_text_list = mall_rating_text_list or (
                    mall_detail_info.get("dsr") or {}
                ).get("mall_rating_text_list")
                service_score = service_score or (
                    mall_detail_info.get("dsr") or {}
                ).get("service_score")
                open_time_label = (
                    open_time_label
                    or (mall_detail_info.get("open_time_label") or {"desc": None})[
                        "desc"
                    ]
                )
                if mall_basic_info and mall_basic_info.get("is_open") == 0:
                    shop_status = "该店铺已失效"
                    flag = -1

        if shop_name and mall_id:
            item = {
                "shop_name": shop_name,
                "mall_star": mall_star,
                "mall_details_label_list": mall_details_label_list,
                "shop_url": shop_url,
                "open_time_label": open_time_label,
                "mall_desc": mall_desc,
                "mall_logo": mall_logo,
                "mall_label_volist": mall_label_volist,
                "mall_rating_text_list": mall_rating_text_list,
                "service_score": service_score,
            }
            if shop_status:
                item["shop_status"] = shop_status
            if flag is not None:
                item["flag"] = flag
            logger.info(f"店铺 {mall_id} 店铺信息 {item}")
            self.to_db.update_smart(TABLE_NET_PDD_SHOP_INFO, item, f"mall_id = '{mall_id}'")

    def parse_mall_combination(self, df_pro):
        if df_pro.empty:
            return df_pro
        mall_df = df_pro[df_pro["base_url"].str.contains("/mall_combination")].copy()
        if not mall_df.empty:
            mall_df_new = mall_df[
                ~mall_df["response"].map(lambda x: x.get("result") is None)
            ]
            mall_df_new.groupby("mall_id", as_index=False).apply(
                lambda df_one: self._parse_one_mall_combination(df_one)
            )
        return df_pro

    def parse_query_mall_detail_info(self, df_pro):
        if df_pro.empty:
            return df_pro
        mall_df = df_pro[
            df_pro["base_url"].str.contains("turing/mall/query_mall_detail_info")
        ].copy()
        mall_df_new = mall_df[~mall_df["response"].map(lambda x: x is None)]
        if not mall_df_new.empty:
            mall_df_new.groupby("mall_id", as_index=False).apply(
                lambda df_one: self._parse_one_mall_detail(df_one)
            )
        return df_pro

    def parse_mall_info(self, df_pro):
        if df_pro.empty:
            return df_pro
        mall_df = df_pro[
            df_pro["base_url"].str.contains("mall/{}/info")
        ].copy()
        mall_df_new = mall_df[~mall_df["response"].map(lambda x: x is None)]
        if not mall_df_new.empty:
            mall_df_new.groupby("mall_id", as_index=False).apply(
                lambda df_one: self._parse_one_mall_info(df_one)
            )
        return df_pro

    def run_by_ids(self, mall_ids):
        logger.info(f"清洗开始处理数据 {mall_ids}")
        df_pro = self.get_datas(mall_ids)
        df_pro = self.parse_query_cat_goods(df_pro)
        df_pro = self.parse_mall_page(df_pro)
        df_pro = self.parse_mall_combination(df_pro)
        df_pro = self.parse_query_mall_detail_info(df_pro)
        df_pro = self.parse_mall_info(df_pro)

    def run(self, time_range=None):
        if time_range is None:
            time_range = datetime.datetime.now().strftime("%Y-%m-%d")
        mall_ids = self.to_db.find(
            f"select distinct mall_id from net_pdd_proxy_log where create_time > '{time_range}'",
            to_json=True,
        )
        mall_ids = [i["mall_id"] for i in mall_ids if i["mall_id"]]
        batch_size = 100
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            for i in range(0, len(mall_ids), batch_size):
                sub_mall_ids = mall_ids[i:i + batch_size]
                executor.submit(self.run_by_ids, sub_mall_ids)


if __name__ == "__main__":
    # CleanToTask().run(time_range=(datetime.datetime.now() - datetime.timedelta(days=25)).strftime("%Y-%m-%d %H:%M:%S"))

    CleanToTask().run_by_ids(
        [
            "100052531",
            # "370926360",
            # "870377347",
            # '545797757'
            # '189480638', '198242747', '211496872', '212119633', '2171036', '219545077', '219964768', '2207723', '220937293', '222691133'
            # '699306148',
            # '22447',
            # '739232799',
            # '473639161',
            # '663131700',
            # '106711797',
            # '321684299',
            # '679673111',
            # '172631443',
            # '816483477', '2227241'
            # "562369447",
            # '109548654', '110227140', '171985035', '216184680', '430780433'
            # '168310280'
            # '120869855',
            # '221896409', '179311172', '210607501'
            # '1000098', '1001136', '100119284', '100120', '100136719', '100169682', '100197', '1002416', '100381594', '100419021', '100499971', '100593089'
        ]
    )
